
package com.bangcommunity.bbframe.mq.producer;

import com.bangcommunity.bbframe.common.base.MainConfigLoader;
import com.bangcommunity.bbframe.common.utils.io.PropertiesUtil;
import com.bangcommunity.bbframe.common.utils.lang.SpiUtils;
import com.bangcommunity.bbframe.common.utils.lang.StringUtils;
import com.bangcommunity.bbframe.dynconfig.DynConfigRegister;
import com.bangcommunity.bbframe.dynconfig.IDynConfigProcessor;
import com.bangcommunity.bbframe.dynconfig.parser.ConfigValueParser;
import com.bangcommunity.bbframe.dynconfig.parser.StringConfigValueParser;
import com.bangcommunity.bbframe.mq.IQtMqFactory;
import com.bangcommunity.bbframe.mq.kafka.producer.KafkaProducerClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class QtProducerClient {

    static Logger logger = LoggerFactory.getLogger(KafkaProducerClient.class);

    private QtProducerClient() {
    }

    private static class QtProducerClientHolder {

        private static IQtProducer instance;
        private static final String mq_config_group = "common";
        private static final String mq_config_dataId = "mqConfig";
        private static String mq = "kafka";
        private static boolean isInit = false;
        private static final String productCode = MainConfigLoader.getInstance().getProductCode();
        private static final String appCode = MainConfigLoader.getInstance().getAppCode();
        static {
            DynConfigRegister.getInstance().buildOnce(productCode, appCode, mq_config_group, mq_config_dataId,
                    new IDynConfigProcessor() {

                        @Override
                        public void procesConfig(String config) {

                            Properties properties = PropertiesUtil.loadFromString(config);
                            IQtProducer iQtProducer = buildProducer(properties);
                            if (null != iQtProducer) {
                                instance = iQtProducer;
                            } else {
                                logger.warn("didnot modify qtmq build client is null:productCode={},appCode={}",
                                        productCode, appCode);
                            }
                        }
                    });
        }

        /**
         */
        private static IQtProducer buildProducer(Properties properties) {

            if (null == properties) {
                properties = new Properties();
            }
            ConfigValueParser<String> configValueParser = new StringConfigValueParser();
            String newMq = configValueParser.parse(properties, "producer", "kafka");
            if (!isInit) {
                IQtMqFactory mqFactory = SpiUtils.getSpiBean(newMq, IQtMqFactory.class);
                if (null != mqFactory) {
                    isInit = true;
                    return mqFactory.builderPruduer();
                } else {
                    throw new RuntimeException("no qtmqfactory spi:" + newMq);
                }
            } else {
                if (StringUtils.equalsIgnoreCase(newMq, mq)) {
                    logger.info("not change mq spi do nothing");
                } else {
                    IQtMqFactory mqFactory = SpiUtils.getSpiBean(newMq, IQtMqFactory.class);
                    if (null != mqFactory) {
                        return mqFactory.builderPruduer();
                    } else {
                        throw new RuntimeException("no qtmqfactory spi:" + newMq);
                    }
                }
            }
            return null;
        }
    }

    public static IQtProducer getQtProducer() {

        IQtProducer instance = QtProducerClientHolder.instance;
        if (null == instance) {
            throw new RuntimeException("build qtmq producer client exception");
        }
        return instance;
    }
}
